package com.example.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class Runner implements CommandLineRunner {

//    public static void main(String[] args) throws Exception {
//        System.out.printf("开始执行任务");
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStream<Tuple2<String, Integer>> dataStreaming = env
//                .socketTextStream("localhost", 9998)
//                .flatMap(new Splitter())
//                .keyBy(0)
//                .timeWindow(Time.seconds(5))
//                .sum(1);
//        dataStreaming.print();
//        // execute program
//        env.execute("Flink Streaming Java API Skeleton");
//    }

    @Override
    public void run(String... args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//		DataStream<Tuple2<String, Integer>> dataStreaming = env
//				.socketTextStream("localhost", 9998)
//				.flatMap(new Splitter())
//				.keyBy(0)
//				.timeWindow(Time.seconds(5))
//				.sum(1);
//		dataStreaming.print();
//		// execute program
//		env.execute("Flink Streaming Java API Skeleton");
        DataStreamSource<String> localhost = env.socketTextStream("localhost", 9998);
        DataStream<Tuple2<String, Integer>> sum =
                localhost.flatMap(new DemoApplication.Splitter()).keyBy(0).sum(1);
        sum.print();
        env.execute();
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for(String word : sentence.split(" ")){
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

}
